/*
 *  Copyright 2013 Utkin Dmitry
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

#include <string>
#include <QDebug>
#include <QString>
#include <QHash>
#include <QTimer>
#include <QMutexLocker>
#include <QTcpSocket>
#include <QHostAddress>
#include <QtEndian>
#include <QDateTime>
#include <QCoreApplication>
#include <QReadWriteLock>
#include <google/protobuf/descriptor.h>
#include <google/protobuf/descriptor.pb.h>
#include <google/protobuf/message.h>
#include "SocketRpcProtocol.h"
#include "protoqtrpc/rpcmessage.pb.h"
#include "protoqtrpc/options.pb.h"
#include "SocketRpcChannel.h"

namespace protoqtrpc
{

struct SocketRpcChannel::ActiveRequest
{
    quint64 rpcId;
    QDateTime expires;
    ::google::protobuf::Message* response;
    ::google::protobuf::RpcController* controller;
    ::google::protobuf::Closure* done;
    bool isMultiresponse;
};

class SocketRpcChannelPrivate
{
public:
    QHostAddress host;
    quint16 port;
    quint64 timeout;
    quint64 lastRpcId;
    QTcpSocket* sock;
    SocketRpcProtocol* rpcProtocol;
    QTimer* timer;
    QHash<quint64, SocketRpcChannel::ActiveRequest*> messages;
    QMutex mutexMessages;

    SocketRpcChannelPrivate():
        port(0), timeout(30000), lastRpcId(0),
        sock(NULL), rpcProtocol(NULL), timer(NULL)
    {
        qRegisterMetaType<const ::google::protobuf::MethodDescriptor*>(
                    "const ::google::protobuf::MethodDescriptor*");
        qRegisterMetaType< ::google::protobuf::RpcController* >(
                    "::google::protobuf::RpcController*");
        qRegisterMetaType<const ::google::protobuf::Message*>(
                    "const ::google::protobuf::Message*");
        qRegisterMetaType< ::google::protobuf::Message* >(
                    "::google::protobuf::Message*");
        qRegisterMetaType< ::google::protobuf::Closure* >(
                    "::google::protobuf::Closure*");
    }
};

SocketRpcChannel::SocketRpcChannel():
    d_ptr(new SocketRpcChannelPrivate)
{
}


SocketRpcChannel::SocketRpcChannel(const QString& hostPort):
    d_ptr(new SocketRpcChannelPrivate)
{
    int pos = hostPort.lastIndexOf(':');
    Q_ASSERT(pos != -1);
    bool ok = false;
    quint16 port = hostPort.mid(pos + 1).toUShort(&ok);
    Q_ASSERT(ok);

    d_ptr->host = hostPort.left(pos);
    d_ptr->port = port;
    moveToThread(this);
    start();
}

SocketRpcChannel::SocketRpcChannel(const QHostAddress& host, quint16 port):
    d_ptr(new SocketRpcChannelPrivate)
{
    d_ptr->host = host;
    d_ptr->port = port;
    moveToThread(this);
    start();
}

SocketRpcChannel::~SocketRpcChannel()
{
    // Wait for all callbacks to return control.
    d_ptr->mutexMessages.lock();

    if (d_ptr->timer && d_ptr->timer->isActive())
    {
        d_ptr->timer->stop();
        delete d_ptr->timer;
    }

    qDeleteAll(d_ptr->messages.values());
    d_ptr->messages.clear();

    // Stop event loop
    quit();
    wait(1000);

    if (d_ptr->sock)
    {
        d_ptr->sock->disconnectFromHost();
        d_ptr->sock->close();
        d_ptr->sock->deleteLater();

        delete d_ptr->rpcProtocol;
    }

    d_ptr->mutexMessages.unlock();
    delete d_ptr;
}

bool SocketRpcChannel::connectToService(const QHostAddress& host, quint16 port)
{
    if (isRunning())
    {
        qWarning() << "Connection already established!";
        return false;
    }

    d_ptr->host = host;
    d_ptr->port = port;
    moveToThread(this);
    start();

    return true;
}

const QHostAddress& SocketRpcChannel::host() const
{
    return d_ptr->host;
}

quint16 SocketRpcChannel::port() const
{
    return d_ptr->port;
}

bool SocketRpcChannel::isConnectedToService() const
{
    return isRunning() && d_ptr->sock->state() == QAbstractSocket::ConnectedState;
}

bool SocketRpcChannel::handleMessage(const RpcMessage* msg, ActiveRequest* req)
{
    // started inside mutex

    switch (msg->type())
    {
    case RpcMessage::RPC_RESPONSE:
    case RpcMessage::RPC_MESSAGE:
        break;

    default:
        qWarning() << "RPC type (" << RpcMessage::RpcType_Name(msg->type()).c_str()
                   << ") handling is not supported";
        return false;
    }

    switch (msg->status())
    {
    case RpcMessage::STATUS_SUCCESS:
        req->response->ParseFromString(msg->payload());
        break;

    case RpcMessage::STATUS_ERROR:
        req->controller->SetFailed(msg->payload());
        break;

    case RpcMessage::STATUS_TIMEOUT:
        req->controller->SetFailed("TIMEOUT");
        break;

    case RpcMessage::STATUS_CANCELLED:
        req->controller->SetFailed("CANCELLED");
        break;
    }

    if (req->done)
        req->done->Run();

    return true;
}

void SocketRpcChannel::cancel(quint64 rpcId)
{
    if (QThread::currentThread() != thread())
    {
        // rerun within thread loop
        QMetaObject::invokeMethod(this, "cancel", Qt::BlockingQueuedConnection,
                                  Q_ARG(quint64, rpcId));
        return;
    }

    QMutexLocker lock(&d_ptr->mutexMessages);

    RpcMessage cancelMsg;
    cancelMsg.set_payload("CANCEL");
    cancelMsg.set_id(rpcId);
    cancelMsg.set_type(RpcMessage::RPC_CANCEL);

    d_ptr->rpcProtocol->sendMessage(cancelMsg);
}

void SocketRpcChannel::updateTimer()
{
    Q_ASSERT(d_ptr->timer);

    QDateTime minTime;

    foreach (ActiveRequest* status, d_ptr->messages)
        if (!status->expires.isNull() &&
                (minTime.isNull() || minTime > status->expires))
            minTime = status->expires;

    if (!minTime.isNull())
    {
        int delay = QDateTime::currentDateTime().secsTo(minTime) * 1000;

        d_ptr->timer->start(qMax(delay, 0));
    }
}

void SocketRpcChannel::run()
{
    d_ptr->sock = new QTcpSocket;
    d_ptr->rpcProtocol = new SocketRpcProtocol(d_ptr->sock);

    connect(d_ptr->sock, SIGNAL(error(QAbstractSocket::SocketError)),
            SLOT(socketError(QAbstractSocket::SocketError)), Qt::DirectConnection);

    connect(d_ptr->rpcProtocol, SIGNAL(messageReceived(protoqtrpc::RpcMessage)),
            SLOT(messageReceived(protoqtrpc::RpcMessage)), Qt::DirectConnection);

    d_ptr->sock->connectToHost(d_ptr->host, d_ptr->port);

    d_ptr->timer = new QTimer;
    d_ptr->timer->setSingleShot(true);
    connect(d_ptr->timer, SIGNAL(timeout()), SLOT(expire()));

    exec();
}

void SocketRpcChannel::messageReceived(const RpcMessage& msg)
{
    QMutexLocker lock(&d_ptr->mutexMessages);

    QHash<quint64, SocketRpcChannel::ActiveRequest*>::iterator itMessage =
            d_ptr->messages.find(msg.id());

    if (itMessage != d_ptr->messages.end())
    {
        ActiveRequest* status = itMessage.value();
        bool remove = (!status->isMultiresponse && msg.type() != RpcMessage::RPC_MESSAGE);

        if (remove)
            d_ptr->messages.erase(itMessage);

        lock.unlock();

        handleMessage(&msg, status);
        if (remove)
            delete status;
    }
}

void SocketRpcChannel::expire()
{
    QMutexLocker lock(&d_ptr->mutexMessages);

    const QDateTime& now = QDateTime::currentDateTime();

    ActiveRequest* status;
    for (QHash<quint64, ActiveRequest*>::iterator it = d_ptr->messages.begin();
         it != d_ptr->messages.end();)
    {
        status = it.value();
        if (status->expires.isNull() || status->expires > now)
        {
            ++it;
            continue;
        }

        status->controller->SetFailed("TIMEOUT");
        status->done->Run();
#ifdef QT_DEBUG
        qDebug() << "Expired callback for " << status->rpcId;
#endif

        it = d_ptr->messages.erase(it);
        delete status;
    }

    updateTimer();
}

void SocketRpcChannel::CallMethod(const ::google::protobuf::MethodDescriptor* method,
                                  ::google::protobuf::RpcController* controller,
                                  const ::google::protobuf::Message* request,
                                  ::google::protobuf::Message* response,
                                  ::google::protobuf::Closure* done)
{
    if (QThread::currentThread() != thread())
    {
        // rerun within thread loop
        QMetaObject::invokeMethod(this, "CallMethod", Qt::BlockingQueuedConnection,
                                  Q_ARG(const ::google::protobuf::MethodDescriptor*, method),
                                  Q_ARG(::google::protobuf::RpcController*, controller),
                                  Q_ARG(const ::google::protobuf::Message*, request),
                                  Q_ARG(::google::protobuf::Message*, response),
                                  Q_ARG(::google::protobuf::Closure*, done));
        return;
    }

    // this is executed from thread loop
    QMutexLocker l(&d_ptr->mutexMessages);

    if (d_ptr->sock->state() != QAbstractSocket::ConnectedState &&
            !d_ptr->sock->waitForConnected(!d_ptr->timeout ? INT_MAX : d_ptr->timeout))
    {
        controller->SetFailed(("Connection to server failed: " +
                               d_ptr->sock->errorString()).toUtf8().constData());
        done->Run();
        return;
    }

#ifdef QT_DEBUG
    qDebug() << "SocketRpcChannel: Sending RPC of type "
             << method->full_name().c_str();
#endif

    RpcMessage msg;
    msg.set_type(RpcMessage::RPC_REQUEST);
    msg.set_service_name(method->service()->full_name());
    msg.set_method_name(method->name());
    msg.set_timeout(d_ptr->timeout);
    msg.set_payload(request->SerializeAsString());


    ActiveRequest* status = new ActiveRequest;
    status->rpcId = ++d_ptr->lastRpcId;
    status->done = done;
    status->response = response;
    status->controller = controller;
    status->isMultiresponse = method->options().GetExtension(protoqtrpc::multiresponse);

    if (d_ptr->timeout && !status->isMultiresponse)
        status->expires = QDateTime::currentDateTime().addMSecs(d_ptr->timeout);

    msg.set_id(status->rpcId);
    d_ptr->messages[status->rpcId] = status;

    controller->NotifyOnCancel(
                ::google::protobuf::NewPermanentCallback(this, &SocketRpcChannel::cancel,
                                                         status->rpcId));

    d_ptr->rpcProtocol->sendMessage(msg);

    updateTimer();
}

void SocketRpcChannel::setTimeout(quint64 timeout)
{
    d_ptr->timeout = timeout;
}

quint64 SocketRpcChannel::timeout() const
{
    return d_ptr->timeout;
}

void SocketRpcChannel::socketError(QAbstractSocket::SocketError err)
{
    if (err == QAbstractSocket::RemoteHostClosedError)
    {
        qDebug() << "Connection closed by server ("
                 << d_ptr->sock->peerAddress().toString() << ")";
    }
    else
    {
        qWarning() << "Error " << err << ": " << d_ptr->sock->errorString();
    }
}


}
